package com.camemax.operators;

import com.camemax.pojo.AdvertisementView;
import com.camemax.pojo.BlackList;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import scala.Tuple2;

/*
* 实现黑名单过滤，当单个userID点击广告次数大于指定值时，触发报警： 主数据流不显示，分流至侧输出流
* */
public class BlackListKeyedProcessFunction extends KeyedProcessFunction<Tuple2<Long, Long>, AdvertisementView, AdvertisementView> {

    private Integer blackListThreshold;

    public BlackListKeyedProcessFunction(Integer threshold){
        this.blackListThreshold = threshold;
    }

    private ValueState<Long> clickCount; // Flink状态： 点击次数,默认0
    private ValueState<Boolean> blackListFlag ; // Flink状态： 广告黑名单用户标志位，默认false

    @Override
    public void open(Configuration parameters) {
        clickCount = getRuntimeContext().getState(
                new ValueStateDescriptor<>("click-count", Long.class));

        blackListFlag = getRuntimeContext().getState(
                new ValueStateDescriptor<>("black-list-flag", Boolean.class));
    }

    @Override
    public void processElement(AdvertisementView value, Context ctx, Collector<AdvertisementView> out) throws Exception {
        // 状态初始化
        if (clickCount.value() == null) {
            clickCount.update(0L);
        }
        if (blackListFlag.value() == null) {
            blackListFlag.update(Boolean.FALSE);
        }

        // 1.1 根据登录次数是否超出黑名单阀值，从而决定是否执行拉黑操作
        if (clickCount.value() >= blackListThreshold) {
            // 1.1.1 如果是未进入黑名单的数据流，则更新黑名单标志位为True，且加入到侧输出流
            if (!blackListFlag.value()) {
                blackListFlag.update(Boolean.TRUE);
                OutputTag<BlackList> blackList = new OutputTag<BlackList>("black-list") {
                };
                Long userId = ctx.getCurrentKey()._1;
                Long advertisementId = ctx.getCurrentKey()._2;
                String message = userId + " click [" + advertisementId + "] over " + blackListThreshold + " times.";
                ctx.output(
                        blackList
                        , new BlackList(userId, advertisementId, message)
                );
            }
            // 1.1.2 如果该数据流已经存在于黑名单了，则啥也不做跳过
            return;
        }

        // 1.2 缓存&更新点击量+1
        clickCount.update(clickCount.value() + 1L);
        out.collect(value);

        // 2. [可选] 清除功能，注册系统时间， 默认第二天00:00清除
//        int dayUnit = 24 * 60 * 60 ;
//        ctx.timerService().registerProcessingTimeTimer(
//                (Long.parseLong(value.getTimestamp()) / dayUnit + 1 ) * 8 * 24 * 60 * 60 // 注册数据流当天的第二天凌晨为WindowEnd Time
//        );
    }
}
